package com.neusoft.databus.core.flume;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import com.google.common.eventbus.AsyncEventBus;
import com.neusoft.bizcore.schedule.service.JobMgrService;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.dto.GatewayDTO;
import com.neusoft.databus.common.dto.ResourceDTO;
import com.neusoft.databus.core.service.GatewayService;
import com.neusoft.databus.core.service.ResourceService;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MetricCollectionProcessor implements MessageCallback<CommandBean> {

    @javax.annotation.Resource(name = "monitorDataEventBus")
    private AsyncEventBus monitorDataEventBus;
    @Autowired
    private GatewayService gatewayService;
    @Autowired
    private ResourceService resourceService;
    @Autowired
    private JobMgrService jobMgrService;
    @Autowired
    StringRedisTemplate template;

    @Getter
    private final AtomicLong successTotal = new AtomicLong(0);
    @Getter
    private final AtomicLong failTotal = new AtomicLong(0);
    // 防止任务频繁重构
    private final Map<String, Long> rebuildJobs = new ConcurrentHashMap<>();

    @Override
    public boolean process(final List<CommandBean> messages) {
        messages.forEach(it -> {
            if (MetricCollectionProcessor.log.isTraceEnabled()) {
                MetricCollectionProcessor.log.trace(JsonUtils.pojoToJson(it));
            }
            it.setEndTime(new Date());
            if (!it.isError()) {
                this.successTotal.addAndGet(1);
                this.monitorDataEventBus.post(it);
            } else {
                this.failTotal.addAndGet(1);
                this.rebuildJob(it);
            }

        });
        return true;
    }

    private void rebuildJob(final CommandBean command) {
        try {
            final GatewayDTO gateway = this.gatewayService.findByCode(command.getGateway());
            String jobName = "";
            if (gateway.isBatched()) {
                jobName = gateway.getCode() + "." + gateway.getCode();
            } else {
                if (command.getResourceBean().size() > 0) {
                    jobName = command.getResourceBean().get(0).getCode() + "." + gateway.getCode();
                }
            }

            if ("".equals(jobName)) {
                return;
            }

            if (gateway.isBatched()) {
                if (this.rebuildJobs.containsKey(jobName)) {
                    if ((System.currentTimeMillis() - this.rebuildJobs.get(jobName)) < 2000) {
                        return;
                    }
                }
                // 说明整个批处理中的资源都没有采集成功，重新生成定时任务
                // 如果agent端将发送的command按照资源拆分成一个个单资源command，说明部分设备采集成功，不重构任务
                if (command.getResourceBean().size() > 1) {
                    final List<ResourceDTO> resources = this.resourceService.findByGateway(gateway.getCode());
                    final Optional<ResourceDTO> optMinIntervalResource =
                            resources.stream()
                                    .min((o1, o2) -> o1.getExecutionCycle() > o2.getExecutionCycle() ? 1 : -1);

                    if (optMinIntervalResource.isPresent()) {
                        this.jobMgrService.rescheduleJob(gateway.getCode(), gateway.getCode(),
                                optMinIntervalResource.get().getExecutionCycle());
                    } else {
                        this.jobMgrService.rescheduleJob(gateway.getCode(), gateway.getCode(), 60 * 3L);
                    }

                    this.rebuildJobs.put(jobName, System.currentTimeMillis());
                }
            } else {
                if (command.getResourceBean().size() > 0) {
                    if (this.rebuildJobs.containsKey(jobName)) {
                        if ((System.currentTimeMillis() - this.rebuildJobs.get(jobName)) < 2000) {
                            return;
                        }
                    }
                    final ResourceDTO resource =
                            this.resourceService.findByCode(command.getResourceBean().get(0).getCode());
                    this.jobMgrService.rescheduleJob(resource.getCode(), gateway.getCode(),
                            resource.getExecutionCycle());
                    this.rebuildJobs.put(jobName, System.currentTimeMillis());
                }

            }
        } catch (final Exception e) {
            MetricCollectionProcessor.log.error(e.getMessage(), e);
        }
    }

}
